package com.miaozhen.datainsight.web.jobMessaging;

import com.fasterxml.jackson.annotation.JsonFormat;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.miaozhen.datainsight.web.enums.IdType;
import com.miaozhen.datainsight.web.util.JSON;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.util.List;

/**
 * @author tzp
 * @date 2017/11/13
 */
@Component
@Slf4j
public class JobSubmitProducer {
    @Autowired
    private KafkaTemplate<Object, Object> kafkaTemplate;

    private static final String KEY_PREFIX = "seg";
    private static final String CONF_KEY_TOPIC_PRODUCER = "";

    public void submitSegment(Long segmentId) {
        JobSubmitMessage msg = new JobSubmitMessage();
        msg.setTaskId(String.valueOf(segmentId));
        try {
            ListenableFuture<SendResult<Object, Object>> future = kafkaTemplate.send(KEY_PREFIX + segmentId,
                    JSON.SNAKE.writeValueAsString(msg));
            future.addCallback(new ListenableFutureCallback<SendResult<Object, Object>>() {
                @Override
                public void onFailure(Throwable ex) {
                    submitError(segmentId, ex);
                }

                @Override
                public void onSuccess(SendResult<Object, Object> result) {
                    log.info("submit job for segment {} success", segmentId);
                }
            });
            Thread.sleep(10000);
        } catch (JsonProcessingException|InterruptedException e) {
            submitError(segmentId, e);
        }
    }

    private void submitError(Long segmentId, Throwable ex) {
        log.error("submit job {} error", segmentId);
    }

    /**
     * 提交任务的消息的payload. see https://gitlab.mzsvn.com/data-platform/data-insight/wikis/CrowdAnalysis
     */
    @Data
    public static class JobSubmitMessage {
        private String taskId;
        private String taskName;
        private Long sampleCount;
        @JsonFormat(shape = JsonFormat.Shape.ANY, pattern = "yyyy-MM-dd")
        private Data startDate;
        @JsonFormat(shape = JsonFormat.Shape.ANY, pattern = "yyyy-MM-dd")
        private Data endDate;

        private List<SegmentPath> segmentPath;

        @Data
        private static class SegmentPath {
            /**
             * id类型的名字 {@link IdType}
             */
            private String uuidType;
            private String filename;
        }
    }
}
